package com.ibm.cps.servlet.starter;

import org.apache.log4j.Logger;

import com.ibm.cps.servlet.TopologyMode;
import com.ibm.cps.servlet.starter.spark.SparkClusterStarter;
import com.ibm.cps.servlet.starter.spark.SparkLocalStarter;
import com.ibm.cps.servlet.starter.storm.StormClusterStarter;
import com.ibm.cps.servlet.starter.storm.StormLocalStarter;
import com.ibm.util.ErrorCode;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;

public class WorkerProcessManager {
	private static final Logger logger = Logger.getLogger(WorkerProcessManager.class);
	
	private final String name;
	private final ProcessMonitor monitor;
	private final UserAppStarter starter;
	
	private WorkerProcessManager(String name, ProcessMonitor monitor, UserAppStarter starter) {
		this.name = name;
		this.monitor = monitor;
		this.starter = starter;
	}
	
	public void start(String tenantId) throws CPSException{
		logger.info("trying to start " + name + " for " + tenantId);
		if(monitor.isAlive(tenantId)){
			logger.info(name + " for " + tenantId + " has already been started");
		}else{
			WorkerProcess process = starter.start(tenantId);
			monitor.registerStarted(tenantId, process);
		}
	}
	
	public void stop(String tenantId) throws CPSException{
		logger.info("trying to stop " + name + " for " + tenantId);
		if(!monitor.isAlive(tenantId)){
			logger.info(name + " for " + tenantId + " has already been stopped");
		}else{
			monitor.stop(tenantId);
		}
	}
	
	public static WorkerProcessManager createWorkerProcessManager(TopologyMode mode) throws CPSException{
		String name = "Topology";
		ProcessMonitor monitor = null;
		UserAppStarter starter = null;
		switch(mode){
		case STORM_LOCAL:
			monitor = new CommandlineProcessMonitor();
			starter = new StormLocalStarter(LocalConfig.TOPOLOGY_START_UP_SHELL_PATH);
			break;
		case STORM_CLUSTER:
			monitor = new CommandlineProcessMonitor();
			starter = new StormClusterStarter(LocalConfig.TOPOLOGY_START_UP_SHELL_PATH);
			throw new CPSException(ErrorCode.NO_IMPLEMENTATION_CODE, "Storm cluster starter is not implemented yet");
		case SPARK_ON_YARN:
			monitor = new YarnApplicationMonitor(LocalConfig.YARN_RESOURCE_MANAGER_URL);
			starter = new SparkClusterStarter(LocalConfig.TOPOLOGY_START_UP_SHELL_PATH);
			break;
		case SPARK_STANDALONE:
			monitor = new CommandlineProcessMonitor();
			starter = new SparkLocalStarter(LocalConfig.TOPOLOGY_START_UP_SHELL_PATH);
		}
		WorkerProcessManager manager = new WorkerProcessManager(name, monitor, starter);
		return manager;
	}
	
	public static WorkerProcessManager createManagerForMessageForwarder(){
		String name = "MessageForwarder";
		ProcessMonitor monitor = new CommandlineProcessMonitor();
		UserAppStarter starter = new MessageForwarderStarter(LocalConfig.FORWARDER_START_UP_SHELL_PATH);
		return new WorkerProcessManager(name, monitor, starter);
	}
}
